package com.shujia.spark.mllib

import org.apache.spark.ml.linalg
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo_ImageMakeData {

    def main(args: Array[String]): Unit = {
      val spark: SparkSession = SparkSession
        .builder()
        .appName("source")
        .master("local[8]")
        .getOrCreate()

      //导入隐式转换
      import spark.implicits._
      //导入spark 所有的函数
      import org.apache.spark.sql.functions._


      //读取图片
      val imageData: DataFrame = spark
        .read
        .format("image")
        .load("F:\\train")

      imageData.show(false)
      imageData.printSchema()

      /**
        * root
        * |-- image: struct (nullable = true)
        * |    |-- origin: string (nullable = true)
        * |    |-- height: integer (nullable = true)
        * |    |-- width: integer (nullable = true)
        * |    |-- nChannels: integer (nullable = true)
        * |    |-- mode: integer (nullable = true)
        * |    |-- data: binary (nullable = true)
        */

      /**
        * 特征工程
        *
        */

      val featuresData: DataFrame = imageData
        .select($"image.origin" as "path", $"image.data" as "data")
        .map(row => {
          val path: String = row.getAs[String]("path")

          //取出文件名
          val name: String = path.split("/").last


          //取出图片的数据
          val data: Array[Byte] = row.getAs[Array[Byte]]("data")

          //处理数据，降噪
          val inData: Array[Double] = data.map(b => {
            val i: Int = b.toInt
            if (i < 0) {
              1.0
            } else {
              0.0
            }
          })

          val vector: linalg.Vector = Vectors.dense(inData)

          (name, vector)
        }).toDF("name", "features")


      //读取标签数据，将图片的数字加上

      val labelData: DataFrame = spark
        .read
        .format("csv")
        .option("sep", " ")
        .schema("name STRING, label DOUBLE")
        .load("F:\\train.txt")


      val dataDF: DataFrame = labelData.join(featuresData, "name")
        .select("label", "features")


      //保存数据
      dataDF
        .write
        .format("libsvm")
        .mode(SaveMode.Overwrite)
        .save("data/imageData")


    }

}
